#include "../../include/thread/thread_pool.h"
#include "../../include/thread/runnable.h"
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>

class RunnablePack
{
public:
  RunnablePack(Runnable *runnable, int priority)
    : runnable_(runnable), priority(priority)
  {
  }

  friend bool operator<(const RunnablePack &p1, const RunnablePack &p2)
  {
    return p1.priority < p2.priority;
  }

  Runnable *runnable_;
  int priority;
};

class ThreadPoolData
{
public:
  ThreadPoolData();
  ~ThreadPoolData();
private:
  friend class ThreadPool;
  std::priority_queue<RunnablePack> thread_queue_;//线程任务队列
  int current_thread_count_{0};
  int max_thread_count_{4};
  std::atomic_int active_thread_count_{0};
  std::atomic_bool init_flag_{false};
  std::mutex mutex_;
  std::condition_variable cv;
  std::vector<std::thread> thread_vector_;
  bool finish_thread_{false};

private:
  void startThread();
  void run();
  bool waitForDone(int msecs = -1);
};
ThreadPoolData::ThreadPoolData() = default;

ThreadPoolData::~ThreadPoolData()
{
  waitForDone();
}
void ThreadPoolData::startThread()
{
  init_flag_.store(true);
  if (current_thread_count_ >= max_thread_count_)
  {
    return;
  }
  for (; current_thread_count_ < max_thread_count_; current_thread_count_++)
  {
    auto thread = std::thread(&ThreadPoolData::run, this);
    thread.detach();
    thread_vector_.push_back(std::move(thread));
  }
}
void ThreadPoolData::run()
{
  while (true)
  {
    std::unique_lock<std::mutex> locker(mutex_);
    cv.wait(locker, [this]
    {
      return !thread_queue_.empty() || finish_thread_;
    });
    if (finish_thread_)
    {
      break;
    }
    Runnable *front = thread_queue_.top().runnable_;
    thread_queue_.pop();
    locker.unlock();
    if (nullptr != front)
    {
      active_thread_count_++;
      if (!front->getRunnableName().empty())
      {
          pthread_setname_np(pthread_self(), front->getRunnableName().c_str());
      }
      front->run();
      if (front->autoDelete())
      {
        delete front;
      }
      active_thread_count_--;
    }
  }
}
bool ThreadPoolData::waitForDone(int msecs)
{
  finish_thread_ = true;
  std::unique_lock<std::mutex> locker(mutex_);
  cv.notify_all();
  std::chrono::time_point<std::chrono::high_resolution_clock> begin = std::chrono::high_resolution_clock::now();
  while (active_thread_count_ > 0)
  {
    if (msecs > 0 && std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::high_resolution_clock::now() - begin).count() >= msecs)
    {
      break;
    }
  }
  return thread_queue_.empty() && 0 == active_thread_count_;
}
ThreadPool::ThreadPool()
  : data_(std::make_unique<ThreadPoolData>())
{
}
ThreadPool::~ThreadPool() = default;

void ThreadPool::start(Runnable *runnable, int priority)
{
  if (!data_->init_flag_.load())
  {
    data_->startThread();
  }
  if (!runnable)
  {
    return;
  }
  std::lock_guard<std::mutex> lock_guard(data_->mutex_);
  data_->thread_queue_.emplace(runnable, priority);
  data_->cv.notify_one();
}
void ThreadPool::clear()
{
  std::lock_guard<std::mutex> lock_guard(data_->mutex_);
  while (!data_->thread_queue_.empty())
  {
    const Runnable *front = data_->thread_queue_.top().runnable_;
    data_->thread_queue_.pop();
    if (nullptr != front && front->autoDelete())
    {
      delete front;
    }
  }
}
void ThreadPool::setMaxThreadCount(int max_thread_count)
{
  data_->max_thread_count_ = max_thread_count;
  data_->startThread();
}
int ThreadPool::maxThreadCount()
{
  return data_->max_thread_count_;
}
bool ThreadPool::waitForDone(int msecs)
{
  return data_->waitForDone(msecs);
}
